package cn.edu.flink.tutorial.broadcast;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.*;
import java.util.Objects;

public class 缓存 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Long> streamSource = env.addSource(累加器.MySourceFunction.getInstance());

        env.registerCachedFile("src/main/resources/hello.txt", "hello");
        streamSource.map(new RichMapFunction<Long, Object>() {
            public File file;

            @Override
            public void open(Configuration parameters) throws Exception {
                file = getRuntimeContext().getDistributedCache().getFile("hello");
            }

            @Override
            public Object map(Long value) throws Exception {
                FileInputStream inputStream = new FileInputStream(file);
                InputStreamReader streamReader = new InputStreamReader(inputStream);
                BufferedReader bufferedReader = new BufferedReader(streamReader);
                for (String line = bufferedReader.readLine(); Objects.nonNull(line); line = bufferedReader.readLine()) {
                    System.out.print(line + " ");
                }
                System.out.println();

                return null;
            }
        });

        env.execute();
    }
}
